package com.jeffse.light.stream.core.function.base;

import com.jeffse.light.stream.core.data.context.*;
import com.jeffse.light.stream.core.data.descriptor.stateDescriptor.ListStateDescriptor;
import com.jeffse.light.stream.core.data.descriptor.stateDescriptor.ValueStateDescriptor;
import com.jeffse.light.stream.core.data.state.ListState;
import com.jeffse.light.stream.core.data.state.ValueState;
import com.jeffse.light.stream.core.function.advance.WindowFunction;
import com.jeffse.light.stream.core.pojo.window.WindowInfo;

import java.util.ArrayList;
import java.util.List;

/**
 *
 * @param <KEY>
 */
public class TimeWindowKeyedProcessFunction<KEY, T, O>  extends KeyedProcessFunction<KEY, T, O> {
    private WindowInfo windowInfo = null;
    private WindowFunction windowFunction = null;
    private ValueState<Long> timestamp;
    private ListState<T> listState;

    public TimeWindowKeyedProcessFunction(WindowInfo windowInfo, WindowFunction windowFunction){
        this.windowInfo = windowInfo;
        this.windowFunction = windowFunction;
    }

    @Override
    public void open(){
        super.open();
        timestamp = getRuntimeContext().getState(new ValueStateDescriptor("timestamp", this));
        listState = getRuntimeContext().getListState(new ListStateDescriptor("listState", this));
    }

    @Override
    public void processElement(ProcessContext context, T value) throws Exception {
        Long time = System.currentTimeMillis() + windowInfo.getNum() * 1000;
        if(timestamp.value() == null){
            getTimerService().registerProcessingTimeTimer(time);
            timestamp.update(time);
        }
        List<T> list = new ArrayList<>();
        if(listState.getList() != null){
            list.addAll(listState.getList());
        }

        list.add(value);
        listState.update(list);
    }

    @Override
    public void onTimer(long timestamp, KeyedContext ctx, ProcessContext sourceContext) throws Exception {
        super.onTimer(timestamp, ctx, sourceContext);
        windowFunction.window(listState.getList(), this);
        this.timestamp.clear();
        listState.clear();
    }
}